package com.wang.sql.operation.window

import com.wang.sql.env.FlinkTableEnv

/**
 * @Desc 滚动窗口与案例场景
 * @Author wang bo
 * @Date 2021/11/2 12:39
 */
object TumbleWindow {
  /**
   * 获取环境
   */
  private val tableEnv=FlinkTableEnv.init_environment()

  /**
   * 用法
   *
   */
  def usage(): Unit ={
    val sql=
      """
        |CREATE TABLE Bid(
        |bidtime STRING,
        |price DECIMAL(10, 2),
        |item STRING,
        |t as TO_TIMESTAMP(bidtime),
        |WATERMARK  FOR t AS t - INTERVAL '1' MINUTES
        |)
        |WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'flink_sql_1',
        |  'properties.group.id'='flink_sql_group_1',
        |  'properties.bootstrap.servers' = 'devcdh1:9092,devcdh2:9092,devcdh3:9092',
        |  'format' = 'json'
        |)
        |""".stripMargin
    tableEnv.executeSql(sql)

    val query=
      """
        |SELECT window_start, window_end, SUM(price) price FROM TABLE(
        |   TUMBLE(
        |     DATA => TABLE Bid,
        |     TIMECOL => DESCRIPTOR(t),
        |     SIZE => INTERVAL '3' MINUTES))
        | GROUP BY window_start, window_end
        |""".stripMargin
    tableEnv.executeSql(query).print()
  }

  /**
   * 场景
   *
   */
  def scene(): Unit ={
    val sql=
      """
        |CREATE TABLE user_clicks (
        |  user_name  VARCHAR ,
        |  click_url VARCHAR ,
        |  ts TIMESTAMP(3),
        |  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND
        |)
        |WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'flink_sql_1',
        |  'properties.group.id'='flink_sql_group_1',
        |  'properties.bootstrap.servers' = 'devcdh1:9092,devcdh2:9092,devcdh3:9092',
        |  'format' = 'json'
        |)
        |""".stripMargin
    tableEnv.executeSql(sql)

    val query=
      """
        |SELECT
        | user_name,
        | count(click_url) as pv,
        | TUMBLE_START(ts, INTERVAL '1' MINUTE) as window_start,
        | TUMBLE_END(ts, INTERVAL '1' MINUTE) as window_end
        | FROM user_clicks
        | GROUP BY TUMBLE(ts, INTERVAL '1' MINUTE), user_name
        |""".stripMargin
    tableEnv.executeSql(query).print()
  }
}
